package net.chenlin.dp.modules.mqtt;

import com.alibaba.excel.util.CollectionUtils;
import lombok.extern.slf4j.Slf4j;
import net.chenlin.dp.common.utils.JSONUtils;
import net.chenlin.dp.modules.newBusiness.dao.DeviceMessageMapper;
import net.chenlin.dp.modules.newBusiness.dao.DeviceOnlineAndOfflineMapper;
import net.chenlin.dp.modules.newBusiness.entity.DeviceStatus;
import net.chenlin.dp.modules.newBusiness.entity.dto.DeviceQueryReplayDto;
import net.chenlin.dp.modules.newBusiness.entity.dto.MonitorDataDto;
import net.chenlin.dp.modules.newBusiness.entity.po.DeviceHistoryStatusPo;
import net.chenlin.dp.modules.newBusiness.entity.po.DeviceMessagePo;
import net.chenlin.dp.modules.newBusiness.entity.po.DeviceQueryReplayPo;
import net.chenlin.dp.modules.newBusiness.entity.vo.TopologyNodeVo;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.time.ZonedDateTime;
import java.time.format.DateTimeFormatter;
import java.util.Date;
import java.util.List;

/**
 * @Author: Yansgm
 * @Description:
 * @Date: 2025-04-27 09:30:25
 */
@Service
@Slf4j
public class DeviceSubscribeSample {

    @Autowired
    private DeviceMessageMapper mapper;

    @Autowired
    private DeviceOnlineAndOfflineMapper andOfflineMapper;

    public void startDevcieSubscribe() {
        List<TopologyNodeVo> list = mapper.getAllDevice();
        if(!CollectionUtils.isEmpty(list)){
            for (TopologyNodeVo vo : list) {
                String dataClientid = vo.getNodeName()+"_data_client";
                String dataTopic = "monitor/"+vo.getNodeName()+"/data";
                start(dataClientid,dataTopic);

                System.out.println("================  设备data订阅完成  ======================");

                String queryReplyClientid = vo.getNodeName()+"/reply";
                String queryReplyTopic = "monitor/"+vo.getNodeName()+"/query_reply";
                start(queryReplyClientid,queryReplyTopic);

                System.out.println("================  设备reply订阅完成  ======================");

                String statusClientid = vo.getNodeName()+"/status";
                String statusTopic = "monitor/"+vo.getNodeName()+"/status";
                start(statusClientid,statusTopic);

                System.out.println("================  设备status订阅完成  ======================");
            }
        }
        log.info("现有在线设备"+list.size()+"台全部启动订阅完成！！！！！");
    }

    public void start(String clientid,String topic) {

        // 获取所有的设备信息

        //EMQ X 默认端口 1883
        // 代理
        String broker = "tcp://1.94.111.92:1883";
        // 主题
        //String TOPIC = "ysm/test";
        int qos = 1;
        //String clientid = "device_start_client";
        String userName = "root";
        String passWord = "Owendeng123";
        try {
            // host为主机名，test为clientid即连接MQTT的客户端ID，一般以客户端唯一标识符表示，MemoryPersistence设置clientid的保存形式，默认为以内存保存
            MqttClient client = new MqttClient(broker, clientid, new MemoryPersistence());
            // MQTT的连接设置
            MqttConnectOptions options = new MqttConnectOptions();
            // 设置是否清空session,这里如果设置为false表示服务器会保留客户端的连接记录，这里设置为true表示每次连接到服务器都以新的身份连接
            options.setCleanSession(true);
            // 设置连接的用户名
            options.setUserName(userName);
            // 设置连接的密码
            options.setPassword(passWord.toCharArray());
            // 设置超时时间 单位为秒
            options.setConnectionTimeout(10);
            // 设置会话心跳时间 单位为秒 服务器会每隔1.5*20秒的时间向客户端发送个消息判断客户端是否在线，但这个方法并没有重连的机制
            options.setKeepAliveInterval(20);
            // 设置回调函数
            client.setCallback(new MqttCallback() {

                @Override
                public void connectionLost(Throwable cause) {
                    System.out.println(clientid+"断开了连接");
                    System.out.println("DeviceSubscribeSample ====== connectionLost");
                }

                @Override
                public void messageArrived(String topic, MqttMessage message) throws ParseException {
                    System.out.println("DeviceSubscribeSample ======监听到来自[" + topic + "]的消息======");
                    String content = new String(message.getPayload());
                    System.out.println("DeviceSubscribeSample message content："+ content);
                    System.out.println("DeviceSubscribeSample ============");
                    if(topic.contains("data")){
                        System.out.println("DeviceSubscribeSample ==== 进去保存data方法");
                        MonitorDataDto dto  = JSONUtils.jsonToBean(content,MonitorDataDto.class);
                        DeviceMessagePo po = new DeviceMessagePo();
                        BeanUtils.copyProperties(dto,po);
                        po.setMonitorId(dto.getMonitor_id());
                        // 获取当前设备所在的台区ID
                        Integer districtId = mapper.getDistrictId(dto.getMonitor_id());
                        po.setDistrictId(districtId);
                        // 解析ISO日期时间字符串
                        ZonedDateTime zonedDateTime = ZonedDateTime.parse(dto.getTimestamp());
                        // 定义目标格式
                        DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
                        // 格式化日期时间
                        String formattedDateTime = zonedDateTime.format(formatter);
                        SimpleDateFormat sdf = new SimpleDateFormat("yyy-MM-dd HH:mm:ss");
                        Date date = sdf.parse(formattedDateTime);
                        po.setReportTime(date);
                        // 解析ISO日期时间字符串
                        ZonedDateTime startzonedDateTime = ZonedDateTime.parse(dto.getMeteringstart());
                        // 格式化日期时间
                        String startformattedDateTime = startzonedDateTime.format(formatter);
                        Date meteringstart = sdf.parse(startformattedDateTime);
                        po.setMeteringstart(meteringstart);

                        // 解析ISO日期时间字符串
                        ZonedDateTime meteringendzonedDateTime = ZonedDateTime.parse(dto.getMeteringend());
                        // 格式化日期时间
                        String meteringendformattedDateTime = meteringendzonedDateTime.format(formatter);
                        Date meteringenddate = sdf.parse(meteringendformattedDateTime);
                        po.setMeteringend(meteringenddate);

                        po.setMeteringA(dto.getMeteringvalue().getMeteringA());
                        po.setMeteringB(dto.getMeteringvalue().getMeteringB());
                        po.setMeteringC(dto.getMeteringvalue().getMeteringC());

                        po.setCurrentA(dto.getCurrentA());
                        po.setCurrentB(dto.getCurrentB());
                        po.setCurrentC(dto.getCurrentC());

                        po.setVoltageA(dto.getVoltagevalue().getVoltageA());
                        po.setVoltageB(dto.getVoltagevalue().getVoltageB());
                        po.setVoltageC(dto.getVoltagevalue().getVoltageC());
                        mapper.insertDeviceMessage(po);
                    }
                    if(topic.contains("query_reply")){
                        System.out.println("DeviceSubscribeSample ==== 进去保存query_reply方法");
                        DeviceQueryReplayDto dto  = JSONUtils.jsonToBean(content,DeviceQueryReplayDto.class);
                        DeviceQueryReplayPo po = new DeviceQueryReplayPo();
                        BeanUtils.copyProperties(dto,po);
                        po.setMonitorId(dto.getMonitor_id());
                        // 解析ISO日期时间字符串
                        ZonedDateTime zonedDateTime = ZonedDateTime.parse(dto.getTimestamp());
                        // 定义目标格式
                        DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
                        // 格式化日期时间
                        String formattedDateTime = zonedDateTime.format(formatter);
                        SimpleDateFormat sdf = new SimpleDateFormat("yyy-MM-dd HH:mm:ss");
                        Date date = sdf.parse(formattedDateTime);
                        po.setTimestamp(date);
                        mapper.insertDeviceQueryReplay(po);
//                        if(dto.getCurrentA().compareTo(BigDecimal.ZERO) != 0 && dto.getVolA().compareTo(BigDecimal.ZERO) != 0){
//                            int i = topic.indexOf("/");
//                            int y = topic.lastIndexOf("/");
//                            String monitorId = topic.substring(i+1, y);
//                            System.out.println("topic 的监测终端回复查询命令 设备 monitorId  == "+monitorId);
//                            // 修改设备为上线状态
//                            mapper.updateOnline(monitorId);
//                        }
                    }
                    if(topic.contains("status")){
                        DeviceStatus dto  = JSONUtils.jsonToBean(content, DeviceStatus.class);
                        System.out.println("DeviceSubscribeSample ==== 进去修改设备status方法");
                        if(dto.getStatus() != null){
                            int i = topic.indexOf("/");
                            int y = topic.lastIndexOf("/");
                            String monitorId = topic.substring(i+1, y);
                            System.out.println("topic 遗嘱消息的 设备 monitorId  == "+monitorId);
                            if(dto.getStatus().equals("connected")){
                                System.out.println("DeviceSubscribeSample ==== 进去修改设备为在线");
                                // 修改设备为在线状态
                                mapper.updateOnline(monitorId);
                                DeviceHistoryStatusPo statusPo = new DeviceHistoryStatusPo();
                                statusPo.setMonitorId(monitorId);
                                statusPo.setOnlineTime(new Date());
                                andOfflineMapper.insertDeviceHistoryStatus(statusPo);

                                // 重新监听对应的topic
                                String dataClientid = monitorId+"_data_client";
                                String dataTopic = "monitor/"+monitorId+"/data";
                                start(dataClientid,dataTopic);
                            }
                            if(dto.getStatus().equals("disconnected")){
                                System.out.println("DeviceSubscribeSample ==== 进去修改设备为离线");
                                // 修改设备为离线状态
                                mapper.updateOffline(monitorId);
                                // 删除历史的查询数据
                                mapper.updateDeviceQueryReplay(monitorId);
                                DeviceHistoryStatusPo statusPo = new DeviceHistoryStatusPo();
                                statusPo.setMonitorId(monitorId);
                                statusPo.setOfflineTime(new Date());
                                andOfflineMapper.insertDeviceHistoryStatus(statusPo);
                                try {
                                    System.out.println("DeviceSubscribeSample === 下面操作为断开连接");
                                    client.disconnect();
                                    client.close();
                                } catch (MqttException e) {
                                    throw new RuntimeException(e);
                                }

                            }
                        }
                    }
                }

                @Override
                public void deliveryComplete(IMqttDeliveryToken token) {
                    System.out.println("deliveryComplete---------"+ token.isComplete());
                }

            });

            // 建立连接
            System.out.println("连接到 broker: " + broker);
            client.connect(options);

            System.out.println("订阅方 -- 连接成功.");
            //订阅消息
            client.subscribe(topic, qos);
            System.out.println("订阅方 开始监听" + topic);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
